package com.sbp.message.proxy;

import com.alibaba.fastjson.JSON;
import com.dap.exception.BusinessException;
import com.dap.exception.PlatformException;
import com.sbp.message.api.common.CommonUtils;
import com.sbp.message.api.common.TopicUtils;
import com.sbp.message.api.entity.*;
import com.sbp.message.entity.ConsumerMessageRecord;
import com.sbp.message.api.service.MessageService;
import com.sbp.message.conf.ActiveMQConfig;
import com.sbp.message.exception.ExceptionConstants;
import com.sbp.message.service.ConsumerMessageRecordService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import javax.jms.TextMessage;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;

public class MessageProxy implements InvocationHandler {
    private static final Logger logger = LoggerFactory.getLogger(MessageProxy.class);

    private ActiveMQConfig activeMQConfig;
    public void setActiveMQConfig(ActiveMQConfig activeMQConfig) {
        this.activeMQConfig = activeMQConfig;
    }

    private MessageService messageService;
    public void setMessageService(MessageService messageService) {
        this.messageService = messageService;
    }


    private ConsumerMessageRecordService consumerMessageRecordService;
    public void setConsumerMessageRecordService(ConsumerMessageRecordService consumerMessageRecordService) {
        this.consumerMessageRecordService = consumerMessageRecordService;
    }

    private ServiceProcessor serviceProcessor;
    public void setServiceProcessor(ServiceProcessor serviceProcessor) {
        this.serviceProcessor = serviceProcessor;
    }

    private ConsumerServicesManager consumerServicesManager;
    public void setConsumerServicesManager(ConsumerServicesManager consumerServicesManager) {
        this.consumerServicesManager = consumerServicesManager;
    }



    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        //如果传进来是一个已实现的具体类（本次演示略过此逻辑)
        if (Object.class.equals(method.getDeclaringClass())) {
            try {
                return method.invoke(this, args);
            } catch (Throwable e) {
                logger.warn("", e);
                throw new PlatformException(ExceptionConstants.MESSAGE_REFLECT_ERROR, e);
            }
            //如果传进来的是一个接口（核心)
        } else {
            return run(method, args);
        }
    }



    // 消费之前, 判断是否需要消费
    private boolean beforeConsume(Message message) {
        ConsumerMessageRecord consumerMessageRecord = consumerMessageRecordService.selectOne(message.getId());

        // 没有本地消费记录, 可以消费
        if (null == consumerMessageRecord)
            return true;

        // 复活
        if (message.retry >= CommonUtils.MAX_RETIES
                && LocalRecordStatus.LocalRecordStatus_succeed.getValue() != consumerMessageRecord.getStatus()
                && LocalRecordStatus.LocalRecordStatus_processing.getValue() != consumerMessageRecord.getStatus()) {
            logger.info("retry message, consumerMessageRecord={}, message=\n{}", JSON.toJSONString(consumerMessageRecord), JSON.toJSONString(message));
            consumerMessageRecordService.delete(message.getId());   // 删除本地业务异常记录
            return true;
        }

        return false;
    }


    /**
     * 返回false代表不需要后续处理 (非持久化消息 或者 已经消费过)
     * */
    private boolean consumeMessage(Message message) {
        MDC.put(com.dap.commons.Constants.TRACE_ID, message.traceId);

        // 非持久化消息, 直接执行
        if (message.getIsPersistence() == 0) {
            ConsumerServicesManager.SunriseServiceAndParam ssap = consumerServicesManager.resolveSunriseServiceAndParam(message);
            if (null == ssap) return false;
            ssap.service.execute(null, ssap.stringInput);
            return false;
        }

        // 消息已经处理过, 不再处理
        if (!beforeConsume(message)) {
            if (logger.isDebugEnabled())
                logger.debug("message already handled, message={}", JSON.toJSONString(message));
            return false;
        }

        ConsumerServicesManager.SunriseServiceAndParam ssap = consumerServicesManager.resolveSunriseServiceAndParam(message);
        if (null == ssap) return false;
        return serviceProcessor.executeServiceAndInsertRecord(message, ssap.service, ssap.stringInput);
    }

    /**
     * 实现接口的核心方法
     */
    public Object run(Method method, Object[] args) throws Exception {
        TextMessage textMessage = (TextMessage) args[0];
        Message message = null;

        boolean consumeResult = true;
        Exception exception = null;
        try {
            String messageJson = textMessage.getText();
            logger.debug(messageJson);
            message = JSON.parseObject(messageJson, Message.class);

            consumeResult = consumeMessage(message);
        } catch (Exception e) {
            logger.warn("message=" + textMessage.getText(), e);
            exception = e;
        }


        // 手动确认消息
        try {
            textMessage.acknowledge();
        } catch (Throwable e) {
            logger.warn("acknowledge failed, message=" + textMessage.getText(), e);
        }


        // 不需要后续处理
        if (!consumeResult) {
            return ExceptionConstants.SUCCESS;
        }


        if (message != null && message.getIsPersistence() != 0) {
            // 尝试正常结束消息
            if (exception == null) {
                try {
                    messageService.tryEndMessageWithConsumeSucceed(
                            message,
                            TopicUtils.leaveOutInvalidChar(activeMQConfig.getAppName()));
                } catch (Exception e) {
                    logger.warn("message={}", textMessage.getText());
                    logger.error("严重错误,必须人工干预", e);
                    throw new PlatformException(ExceptionConstants.MESSAGE_REMOTE_ERROR);
                }
            }
            // 尝试错误结束消息
            else if (hasBusinessExceptionInCause(exception)) {
                try {
                    serviceProcessor.insertConsumerMessageRecord(message, LocalRecordStatus.LocalRecordStatus_businessFailed);
                    messageService.tryEndMessageWithBusinessFailed(
                            message,
                            TopicUtils.leaveOutInvalidChar(activeMQConfig.getAppName()),
                            exception.getMessage(),
                            generateExceptionStackTraceStr(exception));
                } catch (Exception e) {
                    logger.warn("message={}", textMessage.getText());
                    logger.warn("", e);
                    throw new PlatformException(ExceptionConstants.MESSAGE_REMOTE_ERROR);
                }
            }
            // 修改状态为发送中, 等待检查任务检查重发; 忽略sql执行可能出现的异常消息, 只打印; 最后抛出原来的异常
            else {
                try {
                    messageService.updateMessageStatus(
                            message.getId(),
                            MessageStatus.MessageStatus_sending,
                            exception.getMessage(),
                            generateExceptionStackTraceStr(exception));
                } catch (Exception e) {
                    logger.warn("message={}", textMessage.getText());
                    logger.warn("", e);
                }
                throw exception;
            }
        }

        return ExceptionConstants.SUCCESS;
    }

    private static boolean hasBusinessExceptionInCause(Throwable e) {
        while (e != null) {
            if (e instanceof BusinessException) {
                return true;
            }
            e = e.getCause();
        }
        return false;
    }
    private static String generateExceptionStackTraceStr(Throwable e) {
        ByteArrayOutputStream os = new ByteArrayOutputStream();
        PrintStream ps = new PrintStream(os);

        e.printStackTrace(ps);

        try {
            return os.toString("UTF-8");
        } catch (UnsupportedEncodingException e1) {
            return null;
        } finally {
            ps.close();
        }
    }

}